package com.lmm.rpc.core.cluster;

import com.google.gson.Gson;
import com.lmm.rpc.core.exception.RpcException;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.impl.AbandonedConfig;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @title: ZKClusterImpl
 * @Description: zk服务发现中心
 * @Author Lmm
 * @Date: 2022/9/29 16:51
 * @Version 1.0
 *
 * 客户端代理类需要去维护一个ZKClusterImpl和接口名之间的映射   todo  单列bean
 */
@Slf4j
public class ZKClusterImpl<T> extends AbstractCluster<T> {


    public static final int SESSION_TIMEOUT = 3000;

    public static final int ZK_CONNECT_TIMEOUT = 5000;

    public static final byte HEARTBEAT = (byte) 2;
    /**
     * 当前服务列表
     * */
    private final List<RemoteServerInfo> serverList = new CopyOnWriteArrayList<>();

    /**
     * todo 维护一个map    key：value    接口名：serverList
     * */


    /**
     * 调度任务：主要负责服务列表的心跳检测
     * */
    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool ( 1 );
    /**
     * zk 集群
     *
     * */
    private final String ipsAndPorts;
    /**
     * 环境：env\prod\test
     * */
    private final String env;

    public CuratorFramework zkClient;

    public ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    public Lock writeLock = lock.writeLock();
    public Lock readLock = lock.readLock();

    //孩子节点监听缓存
    private PathChildrenCache pathChildrenCache;

    public ZKClusterImpl(String ipsAndPorts, ILoadBalancer loadBalancer, String serviceName,String env, boolean async, int conTimeOut, int soTimeOut, GenericObjectPoolConfig genericObjectPoolConfig, AbandonedConfig abandonedConfig) {
        super(loadBalancer, serviceName, async, conTimeOut, soTimeOut, genericObjectPoolConfig, abandonedConfig);
        this.ipsAndPorts = ipsAndPorts;
        this.env = env;
        //验证合法性ipsAndPorts todo
        initZKClient();
    }

    /**
     * 初始化zk会话
     *
     * */
    private void initZKClient() {
        if (zkClient == null){
            ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
            zkClient = CuratorFrameworkFactory.builder()
                    //"ip:2181,ip2:2181,ip3:2181"
                    .connectString(ipsAndPorts)
                    .retryPolicy(retryPolicy)
                    .connectionTimeoutMs(ZK_CONNECT_TIMEOUT)
                    .sessionTimeoutMs(SESSION_TIMEOUT)
                    .build();
            zkClient.start();
            try {
                zkClient.blockUntilConnected(10, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                log.error("zk connect fail,reason：【{}】",e);
                throw new RuntimeException(e);
            }
            //初始化 服务节点 注册监听器
            initZKNode();
        }
    }

    /**
     * 初始化zk节点信息
     * 只会执行一次
     * */
    private void initZKNode() {
        try{
            //根节点创建
            String envPath = env.startsWith("/") ? env : "/".concat(env);
            if (zkClient.checkExists().forPath(envPath) == null){
                zkClient.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(envPath);
            }

            String servicePath = serviceName.startsWith ( "/" ) ? serviceName : "/".concat ( serviceName );
            if (zkClient.checkExists().forPath(envPath+servicePath) == null){
                zkClient.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        //权限
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(envPath+servicePath);
            }
            dealChildren(envPath+servicePath);//拉取服务并注册心跳事件
            addDataChangeListener(envPath+servicePath);//注册节点数据修改监听事件
            initScheduleTask ();//初始化周期任务:其实就是让调度线程周期性发送心跳包，动态监控服务节点
        }catch (Exception e){
            log.error("zk initial fail ！",e);
            throw new RpcException("zk initial fail ！");
        }
    }

    /**
     * 初始化周期任务：周期性执行心跳
     * 1分钟周期
     * */
    private void initScheduleTask() {
        executor.scheduleWithFixedDelay ( new HeartbeatTask(),60,60,TimeUnit.SECONDS );
    }

    /**
     * 为孩子节点注册监听器（同步的方式）
     * */
    private void addDataChangeListener(String path) {
        try {
            pathChildrenCache = new PathChildrenCache(zkClient, path, true);
            pathChildrenCache.getListenable().addListener(new ChildDataChangeListener());
            //同步初始化cache BUILD_INITIAL_CACHE
            //启动时同步初始化Cache，表示创建Cache后，就从服务器拉取对应的数据进行监听器注册；
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void dealChildren(String path) throws Exception {
        List<String> childPaths = zkClient.getChildren().forPath(path);
        //将所有的子节点信息封装缓存起来
        updateServerList ( childPaths, path);
    }

    private void updateServerList(List<String> childPaths, String path) {
        try {
            writeLock.lock();
            if (serverList.size () != 0) {
                //清空服务列表
                serverList.clear ();
            }
            if (serverPollCache.size() > 0){
                //清空资源池
                for (java.lang.String key : serverPollCache.keySet()) {
                    GenericObjectPool<Channel> pool = serverPollCache.get(key);
                    if (pool!= null)
                        pool.close();
                    serverPollCache.remove(key);

                }
            }

            //拉取注册中心的数据:去zk节点把服务提供者的信息拉取下来，然后注册心跳事件
            for (String childPath : childPaths) {
                String currPath = path.concat("/").concat(childPath);
                try {
                    byte[] bytes = zkClient.getData().forPath(currPath);
                    String dataJson = new String(bytes, StandardCharsets.UTF_8);
                    Gson gson = new Gson();
                    RemoteServerInfo server = gson.fromJson(dataJson, RemoteServerInfo.class);
                    serverList.add(server);
                    //心跳 todo  给当前server注册心跳，监听是否能够提供服务
                } catch (Exception e) {
                    log.error("services poll fail!",e);
                }
            }

        }finally {
            writeLock.unlock();
        }

    }

    @Override
    public void destroy() {
        if (zkClient != null){
            try{
                writeLock.lock();
                log.info("【{}】 ready to shutdown now! ",serviceName);
                //清理资源池
                if (serverPollCache.size() > 0){
                    for (String key : serverPollCache.keySet()) {
                        GenericObjectPool<Channel> pool = serverPollCache.get(key);
                        if (pool != null){
                            destroyGenericObjectPool(pool);
                        }
                        serverPollCache.remove(key);
                    }
                }
            }finally {
                writeLock.unlock();
            }
        }
    }

    @Override
    public RemoteServerInfo get() {
        RemoteServerInfo remoteServer;
        try{
            readLock.lock();
            remoteServer = loadBalancer.select(serverList);
        }finally {
            readLock.unlock();
        }
        return remoteServer;
    }

    @Override
    public ServerObject<io.netty.channel.Channel>  getObjectForRemote() {
        RemoteServerInfo remoteServer = this.get();
        if (remoteServer == null){
            log.info("there is no available servers, serviceName={},ipsAndPorts={},env={}",serviceName,ipsAndPorts,env);
            return null;
        }
        String remoteSeverKey = createMapKey(remoteServer);
        if (serverPollCache.containsKey(remoteSeverKey)){
            return createServerObject(serverPollCache.get(remoteSeverKey),remoteServer);
        }
        //不存在 就创建一个新的资源池
        GenericObjectPool<io.netty.channel.Channel> pool = createGenericObjectPool(remoteServer);
        serverPollCache.put(remoteSeverKey,pool);
        return createServerObject(pool,remoteServer);
    }

    /**
     * 管理remoteServer和对应资源池之间的关系
     * */
    private ServerObject<io.netty.channel.Channel> createServerObject(GenericObjectPool<io.netty.channel.Channel> pool, RemoteServerInfo remoteServer) {
        return new ServerObject<io.netty.channel.Channel>(pool,remoteServer);
    }

    /**
     * 服务动态上下线核心：监听器
     * */
    private class ChildDataChangeListener implements PathChildrenCacheListener{

        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
            try{
                ChildData data = event.getData();
                if (data == null)
                    return;
                String dataJson = new String(data.getData(), StandardCharsets.UTF_8);
                switch (event.getType()){
                    case CHILD_ADDED:
                        addServer(new Gson().fromJson(dataJson, RemoteServerInfo.class));
                        log.info("service:【{}】新服务节点上线，path:{},data:{}",serviceName,data.getPath(),dataJson);
                        break;
                    case CHILD_REMOVED:
                        removeServer(new Gson().fromJson(dataJson, RemoteServerInfo.class));
                        log.info("service:【{}】服务节点下线，path:{},data:{}",serviceName,data.getPath(),dataJson);
                        break;
                    case CHILD_UPDATED:
                        updateServer(new Gson().fromJson(dataJson, RemoteServerInfo.class));
                        log.info("service:【{}】服务节点更新，path:{},data:{}",serviceName,data.getPath(),dataJson);
                        break;
                }
            }catch (Exception e){
                log.error("监听出现异常：",e);
                e.printStackTrace();
            }
        }
    }

    /**
     * 服务上线
     * */
    private void addServer(RemoteServerInfo add) {
        try{
            writeLock.lock();
            serverList.add(add);//为什么这里不填加资源池呢？ 原因：负载拿到改RemoteServerInfo之后，会去获取对应的池资源，没有的话会创建，所以这里不填加也没关系
            //todo 添加心跳服务
        }finally {
            writeLock.unlock();
        }
    }

    /**
     * 服务更新
     * */
    private void updateServer(RemoteServerInfo update) {
        try{
            writeLock.lock();
            if (serverList.size() > 0){
                for (int i = 0; i < serverList.size(); i++) {
                    RemoteServerInfo remoteServer = serverList.get(i);
                    if (remoteServer.getIp().equals(update.getIp())&&remoteServer.getPort().equals(update.getPort())){
                        serverList.set(i,update);
                    }
                }
            }
        }finally {
            writeLock.unlock();
        }
    }

    /**
     * 服务下线
     * */
    private void removeServer(RemoteServerInfo remove) {
        try{
            writeLock.lock();
            if (serverList.size() > 0){
                for (int i = 0; i < serverList.size(); i++) {
                    RemoteServerInfo remoteServer = serverList.get(i);
                    if (remoteServer.getIp().equals(remove.getIp())&&remoteServer.getPort().equals(remove.getPort())){
                        //移除服务
                        serverList.remove(i);
                        String mapKey = createMapKey(remoteServer);
                        //移除服务对应的资源池
                        //清除池化资源，避免服务重新上线时连接channel，出现netty channel未清理干净异常异常:w
                        GenericObjectPool<Channel> channelGenericObjectPool = serverPollCache.get(mapKey);
                        if (channelGenericObjectPool != null){
                            channelGenericObjectPool.clear();
                            channelGenericObjectPool.close();
                            serverPollCache.remove(mapKey);
                        }
                        //移除心跳服务 todo
                    }
                }
            }
        }finally {
            writeLock.unlock();
        }
    }

    private class HeartbeatTask implements Runnable {
        @Override
        public void run() {
            try{
                writeLock.lock();
                //发送心跳服务

            }finally {
                writeLock.unlock();
            }

        }
    }

    public String createMapKey(RemoteServerInfo remoteServer){
        return remoteServer.getIp ().concat ( "-" ).concat (remoteServer.getPort ());
    }
}
